package cn.jbooter.mq.autoconfigure.rocketmq;

import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import cn.jbooter.coms.exceptions.FrameworkException;
import cn.jbooter.ctx.autoconfigure.startup.intf.ApplicationStartupInitializer;
import cn.jbooter.mq.autoconfigure.rocketmq.annotation.RocketConsumer;
import cn.jbooter.mq.autoconfigure.rocketmq.coms.AbstractRocketPushConsumer;
import cn.jbooter.mq.autoconfigure.rocketmq.coms.RocketDefaultProducer;
import cn.jbooter.mq.autoconfigure.rocketmq.coms.RocketTemplate;
import cn.jbooter.mq.autoconfigure.rocketmq.properties.RocketConsumerInitInfo;
import cn.jbooter.mq.autoconfigure.rocketmq.properties.RocketProperties;


/**
 * rocketmq 自动配置
 * 
 * @author HeJian
 *
 */
@Configuration
@ConditionalOnProperty(prefix="spring.rocketmq",name="enabled",havingValue="true")
@EnableConfigurationProperties({
	RocketProperties.class
})
public class RocketMQAutoConfigure {
	
	private static final Logger logger = LoggerFactory.getLogger(RocketMQAutoConfigure.class);
	
	/**
	 * 默认的producer
	 * @return
	 */
	@Bean(name="rocketDefaultProducer",initMethod="init",destroyMethod="destroy")
	public RocketDefaultProducer rocketDefaultProducer() {
		RocketDefaultProducer producer = new RocketDefaultProducer();
		producer.setBeanName("rocketDefaultProducer");
		return producer;
	}
	
	/**
	 * 发送mq消息
	 * @return
	 */
	@Bean
	public RocketTemplate rocketTemplate() {
		RocketTemplate template = new RocketTemplate();
		//设置默认的producer
		template.setDefaultMQProducer(rocketDefaultProducer().getProducer());
		//设置事务producer
		
		return template;
	}
	
	/**
	 * 项目启动时候初始化consumer
	 * @return
	 */
	@Bean
	public ApplicationStartupInitializer rocketmqConsumerInitializer() {
		return new ApplicationStartupInitializer() {
			@Override
			public int getOrder() {
				return 2;
			}
			@Override
			public void initialize(ApplicationContext ctx) {
				Map<String, Object> beans = ctx.getBeansWithAnnotation(RocketConsumer.class);
				if(beans==null || beans.isEmpty()) {
					return;
				}
				for(Map.Entry<String, Object> bean : beans.entrySet()) {
					String beanName = bean.getKey();
					if (!AbstractRocketPushConsumer.class.isAssignableFrom(bean.getValue().getClass())) {
					    throw new FrameworkException(beanName + " - consumer未实现AbstractRocketPushConsumer抽象类");
					}
					RocketConsumer anno = ctx.findAnnotationOnBean(beanName, RocketConsumer.class);
					AbstractRocketPushConsumer consumer = (AbstractRocketPushConsumer) bean.getValue();
					//开始初始化consumer
					try {
						RocketConsumerInitInfo initInfo = new RocketConsumerInitInfo();
						initInfo.setBeanName(beanName);
						initInfo.setConsumeMode(anno.consumeMode());
						initInfo.setConsumerGroup(anno.consumerGroup());
						initInfo.setMessageModel(anno.messageModel());
						initInfo.setNamesrvAddr(anno.namesrvAddr());
						initInfo.setTopic(anno.topic());
						initInfo.setTag(anno.tag());
						consumer.init(initInfo);
					} catch (Exception e) {
						logger.error("初始化rocket consumer {} 失败! 失败原因:{}",beanName,e.getMessage());
						e.printStackTrace();
						throw new FrameworkException(e.getMessage());
					}
					logger.info("{} is ready to subscribe topic : {}", beanName,anno.topic());
				}
			}
		};
	}


}
